packages/core/src/v3/utils/flattenAttributes.ts (1)

262-268: Precompute filtered key lookups to avoid repeated scans

filteredKeys?.includes(key) walks the array on every iteration. When unflattening large attribute maps this balloons to O(n·m). Converting the list to a Set once keeps the loop O(n) without changing behavior.

   const result: Record<string, unknown> = {};
 
-  for (const [key, value] of Object.entries(obj)) {
-    if (filteredKeys?.includes(key)) {
+  const filteredKeySet = filteredKeys ? new Set(filteredKeys) : undefined;
+
+  for (const [key, value] of Object.entries(obj)) {
+    if (filteredKeySet?.has(key)) {
       continue;
     }

TaskEvent_Property_Analysis.md (1)

9-31: Fix table section rows to satisfy MD056

markdownlint (MD056) is firing because these section rows only declare a single cell even though the header defines four columns. GitHub will render them, but our lint pipeline keeps failing. Pad the rows out to four cells (and repeat for the other section dividers in the doc).

-| **Core Identity & Structure**  |
+| **Core Identity & Structure** |  |  |  |

Based on markdownlint-cli2

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsx (1)

305-307: Avoid unsafe type assertions for spanOverrides; narrow the type once.

Replace the local cast at usage with a typed alias near destructuring to keep types sound and remove the as at callsite.

Apply:

-  const { events, duration, rootSpanStatus, rootStartedAt, queuedDuration, overridesBySpanId } =
-    trace;
+  const {
+    events,
+    duration,
+    rootSpanStatus,
+    rootStartedAt,
+    queuedDuration,
+    overridesBySpanId: rawOverridesBySpanId,
+  } = trace;
+  const overridesBySpanId = rawOverridesBySpanId as Record<string, SpanOverride> | undefined;
@@
-  const spanOverrides = selectedSpanId ? overridesBySpanId?.[selectedSpanId] : undefined;
+  const spanOverrides = selectedSpanId ? overridesBySpanId?.[selectedSpanId] : undefined;
@@
-              spanOverrides={spanOverrides as SpanOverride | undefined}
+              spanOverrides={spanOverrides}

Also applies to: 328-329, 378-379

apps/webapp/app/presenters/v3/RunStreamPresenter.server.ts (1)

44-45: Confirm eventEmitter lifecycle is per-subscription to avoid cross-subscriber listener removal.

eventEmitter.removeAllListeners() assumes an instance scoped to this subscription. If shared, scope removals to the "message" event or return an isolated emitter from subscribeToTrace.

apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts (2)

70-77: Prefer Web Streams and CompressionStream over Node streams for isomorphism.

Remix routes should avoid Node-specific streams/zlib. Use ReadableStream and pipeThrough(new CompressionStream("gzip")) for portability.

As per coding guidelines

-import { createGzip } from "zlib";
-import { Readable } from "stream";
+// no Node-specific stream imports

@@
-  // Create a Readable stream from the runEvents array
-  const readable = new Readable({
-    read() {
-      runEvents.forEach((event) => {
-        try {
-          if (!user.admin && event.kind === TaskEventKind.LOG) {
-            // Only return debug logs for admins
-            return;
-          }
-          this.push(formatRunEvent(event) + "\n");
-        } catch {}
-      });
-      this.push(null); // End of stream
-    },
-  });
-
-  // Create a gzip transform stream
-  const gzip = createGzip();
-
-  // Pipe the readable stream into the gzip stream
-  const compressedStream = readable.pipe(gzip);
-
-  // Return the response with the compressed stream
-  return new Response(compressedStream as any, {
+  const stream = new ReadableStream({
+    start(controller) {
+      for (const event of runEvents) {
+        try {
+          if (!user.admin && event.kind === TaskEventKind.LOG) continue;
+          controller.enqueue(new TextEncoder().encode(formatRunEvent(event) + "\n"));
+        } catch {}
+      }
+      controller.close();
+    },
+  });
+
+  const body =
+    typeof CompressionStream !== "undefined"
+      ? stream.pipeThrough(new CompressionStream("gzip"))
+      : stream; // fallback: uncompressed if CompressionStream not available
+
+  return new Response(body, {
     status: 200,
     headers: {
       "Content-Type": "application/octet-stream",
       "Content-Disposition": `attachment; filename="${parsedParams.runParam}.log"`,
-      "Content-Encoding": "gzip",
+      ...(typeof CompressionStream !== "undefined" ? { "Content-Encoding": "gzip" } : {}),
     },
   });

---

125-127: De-duplicate util by reusing shared helper.

Import getDateFromNanoseconds from ~/v3/eventRepository/common.server instead of redefining it here.

Apply in this file:

+import { getDateFromNanoseconds } from "~/v3/eventRepository/common.server";
@@
-function getDateFromNanoseconds(nanoseconds: bigint) {
-  return new Date(Number(nanoseconds) / 1_000_000);
-}

Note: If you adopt the precision fix, update the shared helper accordingly in a follow-up.

apps/webapp/app/routes/otel.v1.logs.ts (1)

22-24: Set protobuf response Content-Type

Return header helps OTLP clients decode properly.

-      return new Response(ExportLogsServiceResponse.encode(exportResponse).finish(), {
-        status: 200,
-      });
+      return new Response(ExportLogsServiceResponse.encode(exportResponse).finish(), {
+        status: 200,
+        headers: { "Content-Type": "application/x-protobuf" },
+      });

apps/webapp/app/v3/services/cancelTaskRunV1.server.ts (1)

115-119: Include store in error log context

Adding the store eases debugging across mixed repositories.

   if (cancelRunEventError) {
     logger.error("[CancelTaskRunServiceV1] Failed to cancel run event", {
       error: cancelRunEventError,
       runId: cancelledTaskRun.id,
+      taskEventStore: cancelledTaskRun.taskEventStore,
     });
   }

apps/webapp/app/routes/otel.v1.traces.ts (1)

22-24: Set protobuf response Content-Type

Mirror OTLP expectations for binary responses.

-      return new Response(ExportTraceServiceResponse.encode(exportResponse).finish(), {
-        status: 200,
-      });
+      return new Response(ExportTraceServiceResponse.encode(exportResponse).finish(), {
+        status: 200,
+        headers: { "Content-Type": "application/x-protobuf" },
+      });

internal-packages/clickhouse/src/client/types.ts (2)

93-98: Document and narrow the unsafe insert API

insertUnsafe bypasses schema validation; add JSDoc warning and prefer Record<string, unknown> over any.

-  insertUnsafe<TRecord extends Record<string, any>>(req: {
+  insertUnsafe<TRecord extends Record<string, unknown>>(req: {
     name: string;
     table: string;
     settings?: ClickHouseSettings;
   }): ClickhouseInsertFunction<TRecord>;

---

4-4: Use type-only import for ClickHouseSettings

Avoids unnecessary runtime import.

-import { ClickHouseSettings } from "@clickhouse/client";
+import type { ClickHouseSettings } from "@clickhouse/client";

internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts (1)

780-805: Propagating taskEventStore in emitted events is consistent; add it to runAttemptStarted too

For symmetry and consumer simplicity, include taskEventStore on runAttemptStarted as well.

   this.$.eventBus.emit("runAttemptStarted", {
     time: new Date(),
     run: {
       id: updatedRun.id,
       status: updatedRun.status,
       createdAt: updatedRun.createdAt,
       updatedAt: updatedRun.updatedAt,
       attemptNumber: nextAttemptNumber,
       baseCostInCents: updatedRun.baseCostInCents,
       executedAt: updatedRun.executedAt ?? undefined,
+      taskEventStore: updatedRun.taskEventStore,
     },
     organization: {
       id: updatedRun.runtimeEnvironment.organizationId,
     },
     project: {
       id: updatedRun.projectId,
     },
     environment: {
       id: updatedRun.runtimeEnvironment.id,
     },
   });

Also applies to: 900-913, 978-1001, 1418-1442, 1560-1583

apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts (1)

22-45: Prefer logger over bare console output

We generally steer clear of console.log in the webapp and lean on our logger utilities so messages flow through the configured transports. Could we swap the raw console call for logger.info (or similar) on the ClickHouse instance? Same refactor would help if we ever need to adjust log levels across environments.

internal-packages/clickhouse/schema/007_add_task_events_v1.sql (2)

47-47: Consider reordering the primary key for typical trace queries

If most queries filter by environment_id and trace_id (optionally with a time window), consider ORDER BY (environment_id, trace_id, start_time). Using toUnixTimestamp(start_time) in ORDER BY can reduce data‑skipping for time‑range scans and loses sub-second ordering. Evaluate query patterns before changing.

---

48-49: TTL setting may retain expired rows longer than expected

SETTINGS ttl_only_drop_parts = 1 drops whole parts only. If rows within a part expire at different times, expired rows may persist until the entire part qualifies, impacting retention and storage. If you need row‑level TTL guarantees, remove this setting (default) to allow merges to purge expired rows.

Would you like me to propose a version with row‑level TTL and merge tuning?

apps/webapp/app/v3/eventRepository/index.server.ts (1)

20-30: Store naming consistency

getEventRepository returns store = "clickhouse" or the actual table name ("taskEvent" | "taskEventPartitioned"). Ensure downstream consumers (callbacks, presenters) expect this mixed namespace ("backend" vs "table"). If they need a backend discriminator, return both backend ("postgres" | "clickhouse") and table ("taskEvent" | "taskEventPartitioned").

I can draft a minimal change returning { repository, backend: "postgres" | "clickhouse", storeTable?: "taskEvent" | "taskEventPartitioned" }.

apps/webapp/app/runEngine/concerns/traceEvents.server.ts (1)

14-22: Remove unused ClickHouse DI to simplify concerns

clickhouseEventRepository is no longer used after introducing getEventRepository. Drop the field and constructor arg to avoid confusion and keep DI minimal.

Apply this diff:

-  private readonly clickhouseEventRepository: ClickhouseEventRepository;
+  // no-op

-  constructor(
-    eventRepository: EventRepository,
-    clickhouseEventRepository: ClickhouseEventRepository
-  ) {
+  constructor(eventRepository: EventRepository) {
     this.eventRepository = eventRepository;
-    this.clickhouseEventRepository = clickhouseEventRepository;
   }

You can also remove the unused imports of ClickhouseEventRepository, flags, FEATURE_FLAG, env, and getTaskEventStore if not needed elsewhere.

apps/webapp/app/env.server.ts (3)

1123-1125: Constrain rollout percent to a sane 0–100 range (int).

Prevents invalid percentages and accidental over-rollout.

-    EVENT_REPOSITORY_CLICKHOUSE_ROLLOUT_PERCENT: z.coerce.number().optional(),
+    EVENT_REPOSITORY_CLICKHOUSE_ROLLOUT_PERCENT: z.coerce.number().int().min(0).max(100).optional(),

---

1121-1122: Add lower bounds to batch size and flush interval.

Avoids zero/negative values that would stall batching.

-    EVENTS_CLICKHOUSE_BATCH_SIZE: z.coerce.number().int().default(1000),
-    EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
+    EVENTS_CLICKHOUSE_BATCH_SIZE: z.coerce.number().int().min(1).default(1000),
+    EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS: z.coerce.number().int().min(1).default(1000),

---

1112-1116: Ensure EVENTS_CLICKHOUSE_URL is required when ClickHouse is used.

If EVENT_REPOSITORY_DEFAULT_STORE is "clickhouse" or rollout > 0, this must be present; today it’s optional. Either enforce in schema (discriminated union/refine) or validate at repository initialization.

Would you like me to add a schema refine that asserts EVENTS_CLICKHOUSE_URL is defined when ClickHouse is active?

apps/webapp/app/v3/services/tracePubSub.server.ts (2)

48-53: Await Redis quit to ensure clean shutdown.

Avoids dangling sockets/listeners on unsubscribe.

    const unsubscribe = async () => {
      await redis.unsubscribe(channel);
-      redis.quit();
+      await redis.quit();
      this._subscriberCount--;
    };

---

10-16: Consider removing or exposing _subscriberCount.

Currently unused; either remove to reduce noise or expose a getter for metrics.

apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts (4)

136-139: insertManyImmediate should bypass batching.

Immediate variant currently just queues normally. If callers rely on sync-ish persistence, write directly and publish.

-  async insertManyImmediate(events: CreateEventInput[]): Promise<void> {
-    this.insertMany(events);
-  }
+  async insertManyImmediate(events: CreateEventInput[]): Promise<void> {
+    const records = events.flatMap((e) => this.createEventToTaskEventV1Input(e));
+    const [err] = await this._clickhouse.taskEvents.insert(records);
+    if (err) throw err;
+    await this.#publishToRedis(records);
+  }

---

118-121: Avoid logging full event payloads at debug level.

Attributes/metadata may contain PII; logging entire batches is risky and noisy.

-      logger.debug("ClickhouseEventRepository.flushBatch", {
-        events,
-      });
+      logger.debug("ClickhouseEventRepository.flushBatch", {
+        eventCount: events.length,
+        sampleTraceId: events[0]?.trace_id,
+      });

---

158-161: TTL is hard-coded to 30 days in many places.

Make retention configurable (env or config param) and pass via ClickhouseEventRepositoryConfig.

Based on learnings

Also applies to: 228-231, 256-259, 288-291, 314-317, 448-450, 586-589, 610-612, 658-661, 699-701, 745-747, 791-793, 835-837

---

1312-1324: Harden metadata parsing.

A single malformed JSON string would throw and break the whole request. Catch and return undefined.

-  #parseMetadata(metadata: string): Record<string, unknown> | undefined {
+  #parseMetadata(metadata: string): Record<string, unknown> | undefined {
     if (!metadata) {
       return undefined;
     }

-    const parsed = JSON.parse(metadata);
+    let parsed: unknown;
+    try {
+      parsed = JSON.parse(metadata);
+    } catch {
+      return undefined;
+    }

apps/webapp/app/v3/eventRepository/common.server.ts (3)

37-41: Simplify JS Date duration calc.

Unreachable typeof check; keep it simple.

-export function calculateDurationFromStartJsDate(startTime: Date, endTime: Date = new Date()) {
-  const $endtime = typeof endTime === "string" ? new Date(endTime) : endTime;
-
-  return ($endtime.getTime() - startTime.getTime()) * 1_000_000;
-}
+export function calculateDurationFromStartJsDate(startTime: Date, endTime: Date = new Date()) {
+  return (endTime.getTime() - startTime.getTime()) * 1_000_000;
+}

---

102-131: Ensure error “message” is a string for CUSTOM/STRING errors.

error.raw may not be a string. Coerce safely to avoid downstream type issues.

-    case "CUSTOM_ERROR": {
-      return {
-        type: "Error",
-        message: error.raw,
-      };
-    }
+    case "CUSTOM_ERROR": {
+      return {
+        type: "Error",
+        message: typeof error.raw === "string" ? error.raw : JSON.stringify(error.raw),
+      };
+    }
@@
-    case "STRING_ERROR": {
-      return {
-        type: "Error",
-        message: error.raw,
-      };
-    }
+    case "STRING_ERROR": {
+      return {
+        type: "Error",
+        message: typeof error.raw === "string" ? error.raw : String(error.raw),
+      };
+    }

Please confirm TaskRunError["raw"] is string in all cases; if not, the above prevents non-string assignment.

---

158-166: Simplify empty-object check.

Small nit for readability.

-export function isEmptyObject(obj: object) {
-  for (var prop in obj) {
-    if (Object.prototype.hasOwnProperty.call(obj, prop)) {
-      return false;
-    }
-  }
-
-  return true;
-}
+export function isEmptyObject(obj: object) {
+  return Object.keys(obj).length === 0;
+}

apps/webapp/app/v3/otlpExporter.server.ts (4)

161-168: Clarify log message (it says “Skipping” but allows through).

Message is misleading.

-        logger.debug("Skipping resource span without trigger attribute", {
+        logger.debug("Allowing resource span without trigger attribute", {
           attributes: resourceSpan.resource?.attributes,
           spans: resourceSpan.scopeSpans.flatMap((scopeSpan) => scopeSpan.spans),
         });

---

36-41: Depend on interfaces, not concrete repos.

Constructor types should be IEventRepository to decouple from implementations.

-  constructor(
-    private readonly _eventRepository: EventRepository,
-    private readonly _clickhouseEventRepository: ClickhouseEventRepository,
+  constructor(
+    private readonly _eventRepository: IEventRepository,
+    private readonly _clickhouseEventRepository: IEventRepository,
     private readonly _verbose: boolean,
     private readonly _spanAttributeValueLengthLimit: number
   ) {

---

764-768: Support Uint8Array for bytes values.

OTLP bytes may arrive as Uint8Array, not Buffer.

-function isBytesValue(value: AnyValue | undefined): value is { bytesValue: Buffer } {
+function isBytesValue(value: AnyValue | undefined): value is { bytesValue: Buffer | Uint8Array } {
   if (!value) return false;
 
-  return Buffer.isBuffer(value.bytesValue);
+  return Buffer.isBuffer(value.bytesValue) || value.bytesValue instanceof Uint8Array;
}

---

770-777: Avoid unnecessary Buffer copy in binaryToHex.

Use Buffer’s toString directly when possible.

-function binaryToHex(buffer: Buffer | string | undefined): string | undefined {
+function binaryToHex(buffer: Buffer | Uint8Array | string | undefined): string | undefined {
   if (!buffer) return undefined;
   if (typeof buffer === "string") return buffer;
 
-  return Buffer.from(Array.from(buffer)).toString("hex");
+  return Buffer.isBuffer(buffer)
+    ? buffer.toString("hex")
+    : Buffer.from(buffer).toString("hex");
}

apps/webapp/app/v3/eventRepository/eventRepository.types.ts (1)

222-223: Prefer unknown over any for metadata.

Improves type safety at call sites.

-  metadata: any; // Used by SpanPresenter for entity processing
+  metadata: unknown; // Used by SpanPresenter for entity processing

apps/webapp/app/v3/eventRepository/originalRunIdCache.server.ts (1)

24-42: Redis-backed store must remain optional

We still create the Redis store even when no connection details are configured, so in dev/test it falls back to 127.0.0.1:6379 and every lookup rejects. Please restore the in-memory store as the baseline and only push a RedisCacheStore when the options include a real host/port/socket/url (keeping the keyPrefix/TLS flags). Otherwise this cache breaks anywhere Redis isn’t provisioned.

apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts (1)

1719-1762: Timezone handling is incorrect in DateTime64→Date conversion

Inputs that carry a Z or ±HH:MM suffix still take the fast path, so we drop the offset and end up shifting times by hours. That breaks trace timelines whenever ClickHouse emits offsets. Please skip the fast path when a timezone indicator is present and apply the parsed offset before constructing the Date.

 export function convertClickhouseDateTime64ToJsDate(date: string): Date {
-  // Fast path for common format: "2025-09-23 12:32:46.130262875" or "2025-09-23 12:32:46"
-  // This avoids the expensive regex for the common case
-  if (date.length >= 19 && date[4] === "-" && date[7] === "-" && date[10] === " ") {
+  const tzIndex =
+    date.indexOf("Z", 19) !== -1
+      ? date.indexOf("Z", 19)
+      : Math.max(date.indexOf("+", 19), date.indexOf("-", 19));
+  const hasTimezone = tzIndex !== -1;
+
+  // Fast path for common format without timezone: "2025-09-23 12:32:46.130262875" or "2025-09-23 12:32:46"
+  if (
+    !hasTimezone &&
+    date.length >= 19 &&
+    date[4] === "-" &&
+    date[7] === "-" &&
+    date[10] === " "
+  ) {
@@
-  const fraction = m[7] ?? ""; // up to 9 digits
+  const fraction = m[7] ?? ""; // up to 9 digits
+  const sign = m[8] as "+" | "-" | undefined;
+  const offH = m[9] ? Number(m[9]) : 0;
+  const offM = m[10] ? Number(m[10]) : 0;
@@
-  return new Date(Date.UTC(year, month - 1, day, hour, minute, second, msPart));
+  let msEpoch = Date.UTC(year, month - 1, day, hour, minute, second, msPart);
+
+  if (sign) {
+    const offsetMinutesSigned = (sign === "+" ? 1 : -1) * (offH * 60 + offM);
+    msEpoch -= offsetMinutesSigned * 60_000;
+  }
+
+  return new Date(msEpoch);
 }

apps/webapp/app/v3/otlpExporter.server.ts (1)

256-260: Stop hard-coding environmentType.

This still forces every log/span into DEVELOPMENT, so prod/test traces sent through ClickHouse will be misclassified and downstream filters/analytics stay wrong. Please derive the value from the resource/log attributes (e.g. SemanticInternalAttributes.ENVIRONMENT_TYPE) or repository/env config, and only fall back to "DEVELOPMENT" as a last resort. Apply the same fix in convertSpansToCreateableEvents, too.

apps/webapp/app/services/authorizationRateLimitMiddleware.server.ts (1)

149-152: Consider using createMemoryStore for consistency.

Unlike other files in this PR (billingCache.ts, fairQueueSelectionStrategy.ts) that adopt the createMemoryStore factory, this file still manually instantiates MemoryStore with a custom frequency of 0.001 (vs. the factory's default 0.01).

If the lower frequency is intentional for rate limiting workloads, consider documenting why. Otherwise, using createMemoryStore(limiterCache?.maxItems ?? 1000) would align with the standardization effort and reduce maintenance.

Option 1: Use factory with default frequency (if 0.001 is not critical):

- const memory = new MemoryStore({
-   persistentMap: new Map(),
-   unstableEvictOnSet: { frequency: 0.001, maxItems: limiterCache?.maxItems ?? 1000 },
- });
+ const memory = createMemoryStore(limiterCache?.maxItems ?? 1000);

Option 2: Use factory with custom frequency (if 0.001 is required):

- const memory = new MemoryStore({
-   persistentMap: new Map(),
-   unstableEvictOnSet: { frequency: 0.001, maxItems: limiterCache?.maxItems ?? 1000 },
- });
+ const memory = createMemoryStore(limiterCache?.maxItems ?? 1000, 0.001);

internal-packages/cache/src/stores/memory.ts (1)

5-13: Consider adding parameter validation.

The factory accepts maxItems and frequency without validation. Consider adding guards for edge cases like non-positive values to prevent runtime surprises.

 export function createMemoryStore(maxItems: number, frequency: number = 0.01) {
+  if (maxItems <= 0) {
+    throw new Error("maxItems must be positive");
+  }
+  if (frequency < 0 || frequency > 1) {
+    throw new Error("frequency must be between 0 and 1");
+  }
   return new MemoryStore({
     persistentMap: new Map(),
     unstableEvictOnSet: {
       frequency,
       maxItems,
     },
   });
 }

apps/webapp/app/services/resourceMonitor.server.ts (1)

6-11: Consider enriching the ResourceMonitor configuration.

The current configuration uses an empty ctx and verbose: false. For better observability:

• ctx: Could include app metadata (e.g., { service: "webapp", env: env.NODE_ENV }).
• verbose: Set to false for production but could be toggled via environment variable for debugging.
• processName: Consider setting to identify logs (e.g., "trigger-webapp").

Example enhancement:

+import { env } from "~/env.server";
+
 function initializeResourceMonitor() {
   return new ResourceMonitor({
-    ctx: {},
-    verbose: false,
+    ctx: { service: "webapp", env: env.NODE_ENV },
+    verbose: env.RESOURCE_MONITOR_VERBOSE === "true",
+    processName: "trigger-webapp",
   });
 }

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts (1)

32-35: Honor the EVENTS_CLICKHOUSE_COMPRESSION_REQUEST flag

Request compression is hardcoded to true, so operators cannot disable it even when the new env flag is set. Please derive the boolean from env.EVENTS_CLICKHOUSE_COMPRESSION_REQUEST (e.g. treat "1", "true" case-insensitive as true, otherwise false) instead of forcing compression on.

Apply this diff:

-    compression: {
-      request: true,
-    },
+    compression: {
+      request: env.EVENTS_CLICKHOUSE_COMPRESSION_REQUEST === "1",
+    },

apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts (1)

179-181: Fix bigint comparison in span event filter

BigInt(event.startTime) throws when event.startTime is already a bigint (which it is), so this filter will explode the first time it runs. Compare against the bigint directly instead of re-wrapping it.

-    return records.filter((r) => BigInt(r.start_time) > BigInt(event.startTime));
+    return records.filter((r) => BigInt(r.start_time) > event.startTime);

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts (2)

158-158: Past concern: Guard attributes against undefined.

A previous review flagged that event.properties may be undefined. Looking at line 354, createEventToTaskEventV1InputAttributes accepts Attributes (non-optional). If CreateEventInput.properties can be undefined, pass event.properties ?? {} or update the helper signature to accept Attributes | undefined.

Based on learnings from past review comments.

---

178-180: Past concern: Redundant BigInt conversion.

event.startTime is already a bigint, so wrapping it in BigInt() is unnecessary. The comparison should be:

-    return records.filter((r) => BigInt(r.start_time) > BigInt(event.startTime));
+    return records.filter((r) => BigInt(r.start_time) > event.startTime);

Based on learnings from past review comments.

apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts (5)

487-502: Improve type safety in setAttribute merging.

The setAttribute implementation uses @ts-ignore twice (lines 496, 499) to bypass type checking when merging attributes. Consider defining a helper function with proper types or using a type assertion that's more specific than @ts-ignore to maintain better type safety.

---

1724-1743: Fast path ignores timezone offsets.

The fast path optimization (lines 1724-1743) only handles the common format "2025-09-23 12:32:46.130262875" but doesn't check for or handle timezone offsets like Z, +HH:MM, or -HH:MM. If ClickHouse returns dates with offsets in this format (unlikely but possible), the fast path will produce incorrect results. Either:

1. Add a check to fall through to regex if offset characters are detected (e.g., check for Z, +, - after position 19), or
2. Document that the fast path assumes no timezone suffix.

   // Fast path for common format: "2025-09-23 12:32:46.130262875" or "2025-09-23 12:32:46"
   // This avoids the expensive regex for the common case
-  if (date.length >= 19 && date[4] === "-" && date[7] === "-" && date[10] === " ") {
+  if (date.length >= 19 && date[4] === "-" && date[7] === "-" && date[10] === " " && 
+      !date.includes("Z") && !date.includes("+") && !date.includes("-", 11)) {

Based on learnings from past review comments.

---

1112-1118: Consider adding error handling for JSON.parse.

#parseAttributes calls JSON.parse without error handling (line 1117). If attributes_text contains malformed JSON (due to corruption or bugs), this will throw and crash the query. Consider wrapping in try-catch and returning {} on parse failure, logging the error for debugging.

 #parseAttributes(attributes_text: string): Record<string, unknown> {
   if (!attributes_text) {
     return {};
   }
-  return JSON.parse(attributes_text) as Record<string, unknown>;
+  try {
+    return JSON.parse(attributes_text) as Record<string, unknown>;
+  } catch (e) {
+    logger.error("Failed to parse attributes_text", { attributes_text, error: e });
+    return {};
+  }
 }

---

1328-1354: Consider adding error handling for JSON.parse in #parseMetadata.

Similar to #parseAttributes, #parseMetadata calls JSON.parse (line 1342) without error handling. Malformed metadata could crash queries. Wrap in try-catch and return undefined on parse failure.

 #parseMetadata(
   metadata: string,
   cache: Map<string, Record<string, unknown>>
 ): Record<string, unknown> | undefined {
   if (!metadata) {
     return undefined;
   }

   const cached = cache.get(metadata);
   if (cached) {
     return cached;
   }

-  const parsed = JSON.parse(metadata);
+  let parsed;
+  try {
+    parsed = JSON.parse(metadata);
+  } catch (e) {
+    logger.error("Failed to parse metadata", { metadata, error: e });
+    return undefined;
+  }

   if (typeof parsed !== "object" || parsed === null) {
     return undefined;
   }

   const result = parsed as Record<string, unknown>;
   cache.set(metadata, result);
   return result;
 }

---

88-102: Load shedding is disabled but isDroppableEvent is defined.

loadSheddingEnabled is set to false (line 97), which means the isDroppableEvent function (lines 98-101) will never be called. If load shedding is not needed, consider removing the isDroppableEvent configuration to simplify the code. If it's intended for future use, add a comment explaining the plan.

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

packages/cli-v3/src/commands/dev.ts (1)

149-154: Add error handling for the rules install wizard.

The error from initiateRulesInstallWizard is still being silently swallowed. The MCP install pattern (lines 130-140) captures and logs errors; this code should follow the same approach to inform users when rules installation fails.

Apply this diff to add error handling:

-    await tryCatch(
-      initiateRulesInstallWizard({
-        manifestPath: options.rulesInstallManifestPath,
-        branch: options.rulesInstallBranch,
-      })
-    );
+    const [rulesError] = await tryCatch(
+      initiateRulesInstallWizard({
+        manifestPath: options.rulesInstallManifestPath,
+        branch: options.rulesInstallBranch,
+      })
+    );
+
+    if (rulesError) {
+      log.error(`Failed to install agent rules: ${rulesError.message}`);
+    }

apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts (1)

99-110: Fix OTEL_RESOURCE_ATTRIBUTES format and merge behavior (spec requires key=value, not JSON)

Currently sets OTEL_RESOURCE_ATTRIBUTES to a JSON string and overwrites any existing value. Build a comma‑separated key=value string, merge with existing OTEL_RESOURCE_ATTRIBUTES, and keep the JSON copy in TRIGGER_OTEL_RESOURCE_ATTRIBUTES for internal use. This was flagged earlier and remains unresolved here.

-  if (taskEventStore) {
-    const resourceAttributes = JSON.stringify({
-      [SemanticInternalAttributes.TASK_EVENT_STORE]: taskEventStore,
-    });
-
-    variables.push(
-      ...[
-        { key: "OTEL_RESOURCE_ATTRIBUTES", value: resourceAttributes },
-        { key: "TRIGGER_OTEL_RESOURCE_ATTRIBUTES", value: resourceAttributes },
-      ]
-    );
-  }
+  if (taskEventStore) {
+    const otelAttr = `${SemanticInternalAttributes.TASK_EVENT_STORE}=${taskEventStore}`;
+    const existing = variables.find((v) => v.key === "OTEL_RESOURCE_ATTRIBUTES")?.value;
+    const merged =
+      existing
+        ? (existing.includes(`${SemanticInternalAttributes.TASK_EVENT_STORE}=`) ? existing : `${existing},${otelAttr}`)
+        : otelAttr;
+
+    variables.push({ key: "OTEL_RESOURCE_ATTRIBUTES", value: merged });
+
+    const triggerResourceAttributes = JSON.stringify({
+      [SemanticInternalAttributes.TASK_EVENT_STORE]: taskEventStore,
+    });
+    variables.push({ key: "TRIGGER_OTEL_RESOURCE_ATTRIBUTES", value: triggerResourceAttributes });
+  }

apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts (1)

24-36: Honor the compression toggle for ClickHouse requests

compression.request is still hard-coded to true, so the new EVENTS_CLICKHOUSE_COMPRESSION_REQUEST flag never disables request compression. That makes the operator control ineffective. Please derive the boolean from the env value (treating undefined as the current default of true) before constructing the client.

-  const clickhouse = new ClickHouse({
+  const shouldCompressRequests =
+    env.EVENTS_CLICKHOUSE_COMPRESSION_REQUEST === undefined
+      ? true
+      : ["1", "true", "TRUE"].includes(
+          String(env.EVENTS_CLICKHOUSE_COMPRESSION_REQUEST)
+        );
+
+  const clickhouse = new ClickHouse({
     url: url.toString(),
     name: "task-events",
     keepAlive: {
       enabled: env.EVENTS_CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
       idleSocketTtl: env.EVENTS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
     },
     logLevel: env.EVENTS_CLICKHOUSE_LOG_LEVEL,
     compression: {
-      request: true,
+      request: shouldCompressRequests,
     },
     maxOpenConnections: env.EVENTS_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
   });

apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts (3)

354-370: Make attributes helper null‑safe and align callers.

Accept undefined and return {} consistently; avoids accidental crashes if a caller ever passes undefined.

-  private createEventToTaskEventV1InputAttributes(attributes: Attributes): Record<string, unknown> {
-    const publicAttributes = removePrivateProperties(attributes);
+  private createEventToTaskEventV1InputAttributes(
+    attributes?: Attributes
+  ): Record<string, unknown> {
+    const publicAttributes = removePrivateProperties(attributes);

---

179-181: Fix bigint comparison when filtering span events.

event.startTime is already a bigint; don’t wrap it in BigInt(). Compare the record value (cast) to the bigint directly.

-    return records.filter((r) => BigInt(r.start_time) > BigInt(event.startTime));
+    return records.filter((r) => BigInt(r.start_time) > event.startTime);

---

1724-1768: DateTime64 → JS Date ignores timezone offsets (fast-path applies when it shouldn’t; fallback doesn’t adjust).

When strings have Z or ±HH:MM, the fast path still runs and the fallback path drops the offset. This shifts times in the UI.

Apply both fixes:

• Only take the fast path when there is no trailing offset/Z.
• In the fallback, parse and subtract the offset minutes (mirror the nanoseconds converter).

 export function convertClickhouseDateTime64ToJsDate(date: string): Date {
-  // Fast path for common format: "2025-09-23 12:32:46.130262875" or "2025-09-23 12:32:46"
-  // This avoids the expensive regex for the common case
-  if (date.length >= 19 && date[4] === "-" && date[7] === "-" && date[10] === " ") {
+  // Fast path for common format without timezone suffix:
+  // "YYYY-MM-DD HH:MM:SS(.fff...)" with no trailing Z or ±HH:MM
+  const hasTzSuffix = date.endsWith("Z") || /[+-]\d{2}:?\d{2}$/.test(date);
+  if (!hasTzSuffix && date.length >= 19 && date[4] === "-" && date[7] === "-" && date[10] === " ") {
     const year = Number(date.substring(0, 4));
     const month = Number(date.substring(5, 7));
     const day = Number(date.substring(8, 10));
     const hour = Number(date.substring(11, 13));
     const minute = Number(date.substring(14, 16));
     const second = Number(date.substring(17, 19));

     // Parse fractional seconds if present
     let ms = 0;
     if (date.length > 20 && date[19] === ".") {
       // Take first 3 digits after decimal (milliseconds), pad if shorter
       const fracStr = date.substring(20, Math.min(23, date.length));
       ms = Number(fracStr.padEnd(3, "0"));
     }

     return new Date(Date.UTC(year, month - 1, day, hour, minute, second, ms));
   }

   // Fallback to regex for other formats (T separator, timezone offsets, etc.)
   const s = date.trim();
   const m = CLICKHOUSE_DATETIME_REGEX.exec(s);
   if (!m) {
     throw new Error(`Invalid ClickHouse DateTime64 string: "${date}"`);
   }

   const year = Number(m[1]);
   const month = Number(m[2]); // 1-12
   const day = Number(m[3]); // 1-31
   const hour = Number(m[4]);
   const minute = Number(m[5]);
   const second = Number(m[6]);
   const fraction = m[7] ?? ""; // up to 9 digits
+  const sign = m[8] as "+" | "-" | undefined;
+  const offH = m[9] ? Number(m[9]) : 0;
+  const offM = m[10] ? Number(m[10]) : 0;

   // Convert fractional seconds to exactly 9 digits (nanoseconds within the second).
   const nsWithinSecond = Number(fraction.padEnd(9, "0")); // 0..999_999_999

   // Split into millisecond part (for Date)
   const msPart = Math.trunc(nsWithinSecond / 1_000_000); // 0..999

-  return new Date(Date.UTC(year, month - 1, day, hour, minute, second, msPart));
+  let msEpoch = Date.UTC(year, month - 1, day, hour, minute, second, msPart);
+  if (sign) {
+    const offsetMinutesSigned = (sign === "+" ? 1 : -1) * (offH * 60 + offM);
+    msEpoch -= offsetMinutesSigned * 60_000;
+  }
+  return new Date(msEpoch);
 }

packages/core/src/v3/serverOnly/resourceMonitor.ts (1)

304-307: Only skip our own PID when scanning.

Re-using the <= process.pid guard still drops every target that started before this monitor, so metrics stay empty. Please revert to skipping only process.pid (as noted earlier) in both the /proc scan and ps fallback.

-        if (processPid <= process.pid) {
+        if (processPid === process.pid) {
           continue;
         }
…
-              // Ignore processes that have a lower PID than our own PID
-              return pid > process.pid;
+              return pid !== process.pid;

Also applies to: 369-371

apps/webapp/app/v3/otlpExporter.server.ts (1)

259-260: Stop hard-coding environmentType.

environmentType remains forced to "DEVELOPMENT", so prod/test data lands in the wrong bucket. Please derive it from the OTLP resource/log attributes (e.g., SemanticInternalAttributes.ENVIRONMENT_TYPE) with a final fallback, and keep both logs & spans in sync.

Also applies to: 347-349

apps/webapp/app/v3/services/expireEnqueuedRun.server.ts (1)

81-82: Move eventRepository instantiation inside the ttl conditional block.

The eventRepository is only used when run.ttl is truthy (line 83). Instantiating it unconditionally at line 81 performs unnecessary work when run.ttl is falsy.

Apply this diff to defer instantiation:

-    const eventRepository = resolveEventRepositoryForStore(run.taskEventStore);
-
     if (run.ttl) {
+      const eventRepository = resolveEventRepositoryForStore(run.taskEventStore);
+
       const [completeExpiredRunEventError] = await tryCatch(

apps/webapp/app/eventLoopMonitor.server.ts (1)

68-69: Consider using structured logging for consistency and observability.

The new console.log statement will not be captured by structured log aggregation systems. Since the file already imports logger (line 7) and uses it for structured logging elsewhere (line 110), consider using logger.warn() or logger.info() here for consistency.

Apply this diff to use structured logging:

-    console.log("🥸  Event loop blocked", { asyncType: cached.type, time });
+    logger.warn("event_loop_blocked", { asyncType: cached.type, time });

This change would align better with the PR's observability goals (storing log data in ClickHouse) and ensure the event-loop-blocked events are captured in your logging infrastructure.

apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts (1)

2-2: Use type‑only import for MachinePreset

Avoid pulling the Zod schema value at runtime; import the type only.

-import { MachinePreset, SemanticInternalAttributes } from "@trigger.dev/core/v3";
+import type { MachinePreset } from "@trigger.dev/core/v3";
+import { SemanticInternalAttributes } from "@trigger.dev/core/v3";

apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts (2)

25-26: Import is fine; consider type‑only import for MachinePreset in this file

Reduce runtime bundle by importing MachinePreset as a type.

-import {
-  AckCallbackResult,
-  MachinePreset,
+import {
+  AckCallbackResult,
   V3ProdTaskRunExecution,
   V3ProdTaskRunExecutionPayload,
   TaskRunError,
   TaskRunErrorCodes,
   TaskRunExecution,
   TaskRunExecutionLazyAttemptPayload,
   TaskRunExecutionResult,
   TaskRunFailedExecutionResult,
   TaskRunSuccessfulExecutionResult,
   parsePacket,
   serverWebsocketMessages,
   SemanticInternalAttributes,
 } from "@trigger.dev/core/v3";
+import type { MachinePreset } from "@trigger.dev/core/v3";

---

2187-2219: De‑duplicate this OTEL attribute injection with a shared helper

Both this file and the route construct the same env vars. Extract a small utility (e.g., addOtelResourceAttributes(vars, attrs)) to centralize kv/JSON building and merging.

apps/webapp/app/v3/services/triggerTask.server.ts (1)

97-101: LGTM: per-store tracing is correctly DI’d

Passing both repositories into DefaultTraceEventsConcern aligns with store-aware routing. To avoid per-call instantiation, consider memoizing the concern as a class field.

 export class TriggerTaskService extends WithRunEngine {
+  private readonly traceEventConcern = new DefaultTraceEventsConcern(
+    eventRepository,
+    clickhouseEventRepository
+  );
   public async call(
@@
   private async callV2(
@@
-    const traceEventConcern = new DefaultTraceEventsConcern(
-      eventRepository,
-      clickhouseEventRepository
-    );
+    const traceEventConcern = this.traceEventConcern;

apps/webapp/app/v3/services/triggerTaskV1.server.ts (1)

729-773: Minor: prefer logger over console for consistency

Optional: replace console.log with logger to keep structured logs and consistent levels.

-      console.log("Failed to get queue name: No task found", {
+      logger.debug("Failed to get queue name: No task found", {
@@
-      console.log("Failed to get queue name: Invalid queue config", {
+      logger.debug("Failed to get queue name: Invalid queue config", {

apps/webapp/app/v3/eventRepository/eventRepository.types.ts (2)

222-222: Consider stronger typing for metadata.

The metadata field is typed as any, which eliminates type safety. Consider using unknown or a more specific type to maintain type safety while allowing flexibility.

Apply this diff to improve type safety:

-  metadata: any; // Used by SpanPresenter for entity processing
+  metadata: unknown; // Used by SpanPresenter for entity processing

Then add proper type guards or validation where metadata is accessed.

---

311-417: Consider using type instead of interface.

The coding guidelines indicate "We usually use types over interfaces" for TypeScript. While interface works here, consider refactoring to a type for consistency.

Apply this diff to align with coding guidelines:

-export interface IEventRepository {
+export type IEventRepository = {
   maximumLiveReloadingSetting: number;
   // Event insertion methods
   insertMany(events: CreateEventInput[]): Promise<void>;
   ...
-}
+};

As per coding guidelines.

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsx (1)

378-378: Remove redundant type assertion.

The type assertion as SpanOverride | undefined is redundant since spanOverrides is already inferred as SpanOverride | undefined from line 328.

Apply this diff:

-            spanOverrides={spanOverrides as SpanOverride | undefined}
+            spanOverrides={spanOverrides}

apps/webapp/app/presenters/v3/RunPresenter.server.ts (2)

10-11: env import becomes redundant after unifying max setting.

If you adopt the change above, you can drop the env import.

- import { env } from "~/env.server";
+ // env import no longer needed

Also applies to: 148-149

---

258-264: Harden overridesBySpanId default.

Fallback traceSummary lacks overridesBySpanId; default to {} to keep shape consistent.

-        overridesBySpanId: traceSummary.overridesBySpanId,
+        overridesBySpanId: traceSummary.overridesBySpanId ?? {},

apps/webapp/app/env.server.ts (1)

1114-1130: Tighten ClickHouse env validation and rollout bounds.

• Make EVENTS_CLICKHOUSE_URL required after fallback (consistent with V4_* pattern).
• Use BoolEnv for compression request.
• Bound rollout percent to 0–100.

-    EVENTS_CLICKHOUSE_URL: z
-      .string()
-      .optional()
-      .transform((v) => v ?? process.env.CLICKHOUSE_URL),
+    EVENTS_CLICKHOUSE_URL: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.CLICKHOUSE_URL)
+      .pipe(z.string().min(1)),
-    EVENTS_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
+    EVENTS_CLICKHOUSE_COMPRESSION_REQUEST: BoolEnv.default(true),
-    EVENT_REPOSITORY_CLICKHOUSE_ROLLOUT_PERCENT: z.coerce.number().optional(),
+    EVENT_REPOSITORY_CLICKHOUSE_ROLLOUT_PERCENT: z.coerce.number().int().min(0).max(100).optional(),

internal-packages/clickhouse/src/client/noop.ts (1)

54-63: queryFast columns type should accept ColumnExpression.

Match the interface and the fast builder.

-  public queryFast<TOut extends Record<string, any>, TParams extends Record<string, any>>(req: {
+  public queryFast<TOut extends Record<string, any>, TParams extends Record<string, any>>(req: {
     name: string;
     query: string;
-    columns: string[];
+    columns: Array<string | ColumnExpression>;
     settings?: ClickHouseSettings;
   }): (params: TParams) => Promise<Result<TOut[], QueryError>> {

internal-packages/clickhouse/src/client/queryBuilder.ts (1)

129-134: Type nit: narrow whereIf condition to boolean.

Small clarity/readability win.

-  whereIf(condition: any, clause: string, params?: QueryParams): this {
+  whereIf(condition: boolean, clause: string, params?: QueryParams): this {

apps/webapp/app/runEngine/concerns/traceEvents.server.ts (1)

16-22: Remove unused ClickhouseEventRepository dependency from this class.

It isn’t referenced after introducing #getEventRepository; leaving it adds noise and DI burden.

-  private readonly clickhouseEventRepository: ClickhouseEventRepository;
...
-  constructor(
-    eventRepository: EventRepository,
-    clickhouseEventRepository: ClickhouseEventRepository
-  ) {
+  constructor(eventRepository: EventRepository) {
     this.eventRepository = eventRepository;
-    this.clickhouseEventRepository = clickhouseEventRepository;
   }

If other constructors still pass the second arg, update call sites accordingly.

internal-packages/clickhouse/src/taskEvents.ts (2)

26-38: Consider typed insert with the existing schema for safer dev-time validation.

You already define TaskEventV1Input; using insert(...) provides shape validation while keeping your JSON settings.

-export function insertTaskEvents(ch: ClickhouseWriter, settings?: ClickHouseSettings) {
-  return ch.insertUnsafe<TaskEventV1Input>({
+export function insertTaskEvents(ch: ClickhouseWriter, settings?: ClickHouseSettings) {
+  return ch.insert(TaskEventV1Input)({
     name: "insertTaskEvents",
     table: "trigger_dev.task_events_v1",
     settings: {
       enable_json_type: 1,
       type_json_skip_duplicated_paths: 1,
       input_format_json_throw_on_bad_escape_sequence: 0,
       input_format_json_use_string_type_for_ambiguous_paths_in_named_tuples_inference_from_objects: 1,
       ...settings,
     },
   });
 }

If perf is a concern, keep insertUnsafe; otherwise the typed path can catch regressions early.

---

55-70: Align operation name with the function (“getTraceSummary”) for clearer tracing/metrics.

Purely cosmetic but helps observability.

-    name: "getTraceEvents",
+    name: "getTraceSummary",

internal-packages/clickhouse/src/client/types.ts (1)

4-7: Use type-only imports to avoid unnecessary runtime dependencies and potential cycles.

-import { ClickHouseSettings } from "@clickhouse/client";
+import type { ClickHouseSettings } from "@clickhouse/client";
...
-import { ClickhouseQueryBuilder, ClickhouseQueryFastBuilder } from "./queryBuilder.js";
+import type { ClickhouseQueryBuilder, ClickhouseQueryFastBuilder } from "./queryBuilder.js";

apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts (3)

315-316: Avoid double unflattening on span event metadata.

spanEvent.properties was already unflattened by parseEventsField; stringify directly.

-      metadata: JSON.stringify(unflattenAttributes(spanEvent.properties as Attributes)),
+      metadata: JSON.stringify(spanEvent.properties),

---

98-101: Comment/code mismatch about load shedding drop policy.

Comment says “drop LOG events” but code only drops DEBUG_EVENT. Align the comment to prevent confusion.

-        // Only drop LOG events during load shedding
+        // Only drop debug log events during load shedding
         return event.kind === "DEBUG_EVENT";

---

1664-1665: Docstring example implies a timezone shift that doesn’t happen.

toISOString() is UTC; replacing “T”/removing “Z” does not add +1h. Fix the example.

-  // 2024-11-06T20:37:00.123Z -> 2024-11-06 21:37:00.123
+  // 2024-11-06T20:37:00.123Z -> 2024-11-06 20:37:00.123

apps/webapp/app/v3/services/tracePubSub.server.ts (1)

43-47: Propagate and log Redis subscriber errors; harden unsubscribe.

Surface errors to consumers and avoid silent failures; ensure quit() is awaited safely.

-    // Define the message handler.
-    redis.on("message", (_, message) => {
-      eventEmitter.emit("message", message);
-    });
+    // Define handlers.
+    redis.on("message", (_, message) => eventEmitter.emit("message", message));
+    redis.on("error", (err) => eventEmitter.emit("error", err));
+    redis.on("end", () => eventEmitter.emit("end"));
+    redis.on("reconnecting", () => eventEmitter.emit("reconnecting"));
@@
-      await redis.unsubscribe(channel);
-      redis.quit();
+      try {
+        await redis.unsubscribe(channel);
+      } finally {
+        await redis.quit();
+      }
       this._subscriberCount--;

Also applies to: 49-53

internal-packages/clickhouse/src/client/client.ts (2)

312-326: Guard fast-path hydration against column/row length mismatches.

If the SELECT shape diverges from req.columns, silently misaligned rows will be produced.

-          for (const row of rows) {
+          for (const row of rows) {
             const rowData = row.json();
+            if (!Array.isArray(rowData) || rowData.length !== req.columns.length) {
+              return [
+                new QueryError(
+                  `Column count mismatch: expected ${req.columns.length}, got ${Array.isArray(rowData) ? rowData.length : "non-array"}`,
+                  { query: req.query }
+                ),
+                null,
+              ] as const;
+            }

---

232-270: queryFast: add optional per-column mappers for hot paths. Allow queryBuilderFast to accept an optional ColumnMapper API to enforce per-column type coercions for existing builders (e.g., span_id, parent_span_id in getTraceEvents and getTaskEventDetailedSummary) without impacting performance.